/*-------------------------------------------------------------------------
 *
 * sharedtuplestore.c
 *	  在后端之间共享元组的简单机制。
 *
 * 此模块包含一个共享的临时元组存储机制，提供
 * tuplestore.c功能的并行感知子集。多个后端
 * 可以写入SharedTuplestore，然后多个后端可以随后扫描
 * 存储的元组。当前唯一支持的扫描类型是并行扫描，
 * 每个后端读取所写的元组的任意子集。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, 加利福尼亚大学董事会
 *
 * IDENTIFICATION
 *	  src/backend/utils/sort/sharedtuplestore.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/htup.h"
#include "access/htup_details.h"
#include "miscadmin.h"
#include "storage/buffile.h"
#include "storage/lwlock.h"
#include "storage/sharedfileset.h"
#include "utils/sharedtuplestore.h"

/*
 * 块的大小，以页为单位。这个值是根据HASH_CHUNK的大小任意设定的，
 * 以便并行哈希以大致相同的速率获取新的元组块
 * 与分配新的内存块进行插入的速率相匹配。
 */
#define STS_CHUNK_PAGES 4
#define STS_CHUNK_HEADER_SIZE offsetof(SharedTuplestoreChunk, data)
#define STS_CHUNK_DATA_SIZE (STS_CHUNK_PAGES * BLCKSZ - STS_CHUNK_HEADER_SIZE)

/* 写入磁盘的块。 */
typedef struct SharedTuplestoreChunk
{
	int			ntuples;		/* 此块中的元组数量。 */
	int			overflow;		/* 如果溢出，包括这个在内有多少个？ */
	char		data[FLEXIBLE_ARRAY_MEMBER];
} SharedTuplestoreChunk;

/* 每个参与者的共享状态。 */
typedef struct SharedTuplestoreParticipant
{
	LWLock		lock;
	BlockNumber read_page;		/* 下一个读取的页面编号。 */
	BlockNumber npages;			/* 写入的页数。 */
	bool		writing;		/* 仅用于断言。 */
} SharedTuplestoreParticipant;

/* 生活在共享内存中的控制对象。 */
struct SharedTuplestore
{
	int			nparticipants;	/* 可以写入的参与者数量。 */
	int			flags;			/* 来自SHARED_TUPLESTORE_XXX的标志位 */
	size_t		meta_data_size; /* 每个元组头的大小。 */
	char		name[NAMEDATALEN];	/* 这个元组存储的名称。 */

	/* 后面跟着每个参与者的共享状态。 */
	SharedTuplestoreParticipant participants[FLEXIBLE_ARRAY_MEMBER];
};

/* 生活在后端本地内存中的每个参与者状态。 */
struct SharedTuplestoreAccessor
{
	int			participant;	/* 我的参与者编号。 */
	SharedTuplestore *sts;		/* 共享状态。 */
	SharedFileSet *fileset;		/* 持有文件的SharedFileSet。 */
	MemoryContext context;		/* 缓冲区的内存上下文。 */

	/* 读取的状态。 */
	int			read_participant;	/* 当前读取的参与者。 */
	BufFile    *read_file;		/* 当前读取的文件。 */
	int			read_ntuples_available; /* 块中的元组数量。 */
	int			read_ntuples;	/* 我们已经从块中读取了多少个元组？ */
	size_t		read_bytes;		/* 我们已经从块中读取了多少字节？ */
	char	   *read_buffer;	/* 加载元组的缓冲区。 */
	size_t		read_buffer_size;
	BlockNumber read_next_page; /* 我们考虑读取的最低块。 */

	/* 写入的状态。 */
	SharedTuplestoreChunk *write_chunk; /* 写入的缓冲区。 */
	BufFile    *write_file;		/* 当前写入的文件。 */
	BlockNumber write_page;		/* 下一个写入的页面。 */
	char	   *write_pointer;	/* 当前块内的写入指针。 */
	char	   *write_end;		/* 超过当前块末尾一个的位置。 */
};

static void fc_sts_filename(char *fc_name, SharedTuplestoreAccessor *fc_accessor,
						 int fc_participant);

/*
 * 返回需要的共享内存量以容纳给定数量参与者的SharedTuplestore。
 */
size_t sts_estimate(int fc_participants)
{
	return offsetof(SharedTuplestore, participants) +
		sizeof(SharedTuplestoreParticipant) * fc_participants;
}

/*
 * 在现有共享内存中初始化SharedTuplestore。必须有
 * sts_estimate(participants)字节的空间。如果标志包括值
 * SHARED_TUPLESTORE_SINGLE_PASS，未来可能更积极地删除文件
 * （但这尚未实现）。
 *
 * 存储的元组可以选择性地携带一段固定大小的元数据，
 * 将与元组一起检索。这对于在多批哈希连接中使用的哈希值很有用，
 * 但也可能有其他应用。
 *
 * 调用者必须提供一个SharedFileSet，它本质上是一个目录，
 * 将自动清理，并且一个在相同的SharedFileSet中创建的所有SharedTuplestore中
 * 必须是唯一的名称。
 */
SharedTuplestoreAccessor *
sts_initialize(SharedTuplestore *fc_sts, int fc_participants,
			   int fc_my_participant_number,
			   size_t fc_meta_data_size,
			   int fc_flags,
			   SharedFileSet *fc_fileset,
			   const char *fc_name)
{
	SharedTuplestoreAccessor *fc_accessor;
	int			fc_i;

	Assert(fc_my_participant_number < fc_participants);

	fc_sts->nparticipants = fc_participants;
	fc_sts->meta_data_size = fc_meta_data_size;
	fc_sts->flags = fc_flags;

	if (strlen(fc_name) > sizeof(fc_sts->name) - 1)
		elog(ERROR, "SharedTuplestore name too long");
	strcpy(fc_sts->name, fc_name);

	/*
	 * 限制元数据，使其与元组大小始终适合单一块。
	 * sts_puttuple()和sts_read_tuple()可以支持情景，
	 * 如果情况不是这样，但目前不需要。如果是这样，
	 * 元数据大小可能也应该可变。
	 */
	if (fc_meta_data_size + sizeof(uint32) >= STS_CHUNK_DATA_SIZE)
		elog(ERROR, "meta-data too long");

	for (fc_i = 0; fc_i < fc_participants; ++fc_i)
	{
		LWLockInitialize(&fc_sts->participants[fc_i].lock,
						 LWTRANCHE_SHARED_TUPLESTORE);
		fc_sts->participants[fc_i].read_page = 0;
		fc_sts->participants[fc_i].npages = 0;
		fc_sts->participants[fc_i].writing = false;
	}

	fc_accessor = palloc0(sizeof(SharedTuplestoreAccessor));
	fc_accessor->participant = fc_my_participant_number;
	fc_accessor->sts = fc_sts;
	fc_accessor->fileset = fc_fileset;
	fc_accessor->context = CurrentMemoryContext;

	return fc_accessor;
}

/*
 * 附加到由另一个后端初始化的SharedTuplestore，
 * 以便该后端可以读取和写入元组。
 */
SharedTuplestoreAccessor *
sts_attach(SharedTuplestore *fc_sts,
		   int fc_my_participant_number,
		   SharedFileSet *fc_fileset)
{
	SharedTuplestoreAccessor *fc_accessor;

	Assert(fc_my_participant_number < fc_sts->nparticipants);

	fc_accessor = palloc0(sizeof(SharedTuplestoreAccessor));
	fc_accessor->participant = fc_my_participant_number;
	fc_accessor->sts = fc_sts;
	fc_accessor->fileset = fc_fileset;
	fc_accessor->context = CurrentMemoryContext;

	return fc_accessor;
}

static void fc_sts_flush_chunk(SharedTuplestoreAccessor *fc_accessor)
{
	size_t		fc_size;

	fc_size = STS_CHUNK_PAGES * BLCKSZ;
	BufFileWrite(fc_accessor->write_file, fc_accessor->write_chunk, fc_size);
	memset(fc_accessor->write_chunk, 0, fc_size);
	fc_accessor->write_pointer = &fc_accessor->write_chunk->data[0];
	fc_accessor->sts->participants[fc_accessor->participant].npages +=
		STS_CHUNK_PAGES;
}

/*
 * 完成写入元组。这必须由所有在任何后端开始读取之前已经写入数据的后端调用。
 */
void sts_end_write(SharedTuplestoreAccessor *fc_accessor)
{
	if (fc_accessor->write_file != NULL)
	{
		fc_sts_flush_chunk(fc_accessor);
		BufFileClose(fc_accessor->write_file);
		pfree(fc_accessor->write_chunk);
		fc_accessor->write_chunk = NULL;
		fc_accessor->write_file = NULL;
		fc_accessor->sts->participants[fc_accessor->participant].writing = false;
	}
}

/*
 * 准备重新扫描。只有一个参与者必须调用此函数。返回后，所有参与者可以调用 sts_begin_parallel_scan()，然后循环调用 sts_parallel_scan_next()。此函数不得与扫描并发调用，避免这种情况的同步由调用者负责。
 */
void sts_reinitialize(SharedTuplestoreAccessor *fc_accessor)
{
	int			fc_i;

	/*
	 * 重置所有参与者文件的共享读取头。并将初始块大小设置为最小值（该大小的任何增加将记录在 chunk_expansion_log 中）。
	 */
	for (fc_i = 0; fc_i < fc_accessor->sts->nparticipants; ++fc_i)
	{
		fc_accessor->sts->participants[fc_i].read_page = 0;
	}
}

/*
 * 开始并行扫描内容。
 */
void sts_begin_parallel_scan(SharedTuplestoreAccessor *fc_accessor)
{
	int			fc_i PG_USED_FOR_ASSERTS_ONLY;

	/* 结束任何正在进行的现有扫描。 */
	sts_end_parallel_scan(fc_accessor);

	/*
	 * 任何可能向这个共享元组存储写入数据的后端都必须调用 sts_end_write()，以便所有缓冲区被刷新，并且文件停止增长。
	 */
	for (fc_i = 0; fc_i < fc_accessor->sts->nparticipants; ++fc_i)
		Assert(!fc_accessor->sts->participants[fc_i].writing);

	/*
	 * 我们将首先读取这个后端写入的文件。这样可能会有一些缓存局部性的优势。
	 */
	fc_accessor->read_participant = fc_accessor->participant;
	fc_accessor->read_file = NULL;
	fc_accessor->read_next_page = 0;
}

/*
 * 完成并行扫描，释放相关的后端本地资源。
 */
void sts_end_parallel_scan(SharedTuplestoreAccessor *fc_accessor)
{
	/*
	 * 如果 SHARED_TUPLESTORE_SINGLE_PASS，我们可以在这里删除所有文件，但我们可能需要当前并行扫描器的引用计数，以便我们可以安全地在引用计数达到零时进行删除。
	 */
	if (fc_accessor->read_file != NULL)
	{
		BufFileClose(fc_accessor->read_file);
		fc_accessor->read_file = NULL;
	}
}

/*
 * 写入一个元组。如果提供了元数据大小给 sts_initialize，则必须提供指向该大小的元数据的指针。
 */
void sts_puttuple(SharedTuplestoreAccessor *fc_accessor, void *fc_meta_data,
			 MinimalTuple fc_tuple)
{
	size_t		fc_size;

	/* 我们有自己的文件了吗？ */
	if (fc_accessor->write_file == NULL)
	{
		SharedTuplestoreParticipant *fc_participant;
		char		fc_name[MAXPGPATH];

		/* 创建一个。只有这个后端将写入它。 */
		fc_sts_filename(fc_name, fc_accessor, fc_accessor->participant);
		fc_accessor->write_file =
			BufFileCreateFileSet(&fc_accessor->fileset->fs, fc_name);

		/* 设置这个后端文件的共享状态。 */
		fc_participant = &fc_accessor->sts->participants[fc_accessor->participant];
		fc_participant->writing = true;	/* 仅供断言使用 */
	}

	/* 我们有空间吗？ */
	fc_size = fc_accessor->sts->meta_data_size + fc_tuple->t_len;
	if (fc_accessor->write_pointer + fc_size > fc_accessor->write_end)
	{
		if (fc_accessor->write_chunk == NULL)
		{
			/* 第一次通过。分配块。 */
			fc_accessor->write_chunk = (SharedTuplestoreChunk *)
				MemoryContextAllocZero(fc_accessor->context,
									   STS_CHUNK_PAGES * BLCKSZ);
			fc_accessor->write_chunk->ntuples = 0;
			fc_accessor->write_pointer = &fc_accessor->write_chunk->data[0];
			fc_accessor->write_end = (char *)
				fc_accessor->write_chunk + STS_CHUNK_PAGES * BLCKSZ;
		}
		else
		{
			/* 看看刷新是否有帮助。 */
			fc_sts_flush_chunk(fc_accessor);
		}

		/* 在巨大元组的情况下，这可能仍然不够。 */
		if (fc_accessor->write_pointer + fc_size > fc_accessor->write_end)
		{
			size_t		fc_written;

			/*
			 * 我们将写入超大元组的开头，然后在一定数量的“溢出”块中写入其余部分。
			 *
			 * sts_initialize() 验证元组的大小 + 元数据始终适合一个块。由于上面块已被刷新，我们可以确保有块的所有可用空间可用。
			 */
			Assert(fc_accessor->write_pointer + fc_accessor->sts->meta_data_size +
				   sizeof(uint32) < fc_accessor->write_end);

			/* 将元数据作为一个块写入。 */
			if (fc_accessor->sts->meta_data_size > 0)
				memcpy(fc_accessor->write_pointer, fc_meta_data,
					   fc_accessor->sts->meta_data_size);

			/*
			 * 写入尽可能多的元组内容。这包括元组大小的开头部分。
			 */
			fc_written = fc_accessor->write_end - fc_accessor->write_pointer -
				fc_accessor->sts->meta_data_size;
			memcpy(fc_accessor->write_pointer + fc_accessor->sts->meta_data_size,
				   fc_tuple, fc_written);
			++fc_accessor->write_chunk->ntuples;
			fc_size -= fc_accessor->sts->meta_data_size;
			fc_size -= fc_written;
			/* 现在为其余部分写入所需的尽可能多的溢出块。 */
			while (fc_size > 0)
			{
				size_t		fc_written_this_chunk;

				fc_sts_flush_chunk(fc_accessor);

				/*
				 * 还有多少个溢出块？这将允许读者一次跳过所有块，而不是逐个读取。
				 */
				fc_accessor->write_chunk->overflow = (fc_size + STS_CHUNK_DATA_SIZE - 1) /
					STS_CHUNK_DATA_SIZE;
				fc_written_this_chunk =
					Min(fc_accessor->write_end - fc_accessor->write_pointer, fc_size);
				memcpy(fc_accessor->write_pointer, (char *) fc_tuple + fc_written,
					   fc_written_this_chunk);
				fc_accessor->write_pointer += fc_written_this_chunk;
				fc_size -= fc_written_this_chunk;
				fc_written += fc_written_this_chunk;
			}
			return;
		}
	}

	/* 将元数据和元组复制到缓冲区。 */
	if (fc_accessor->sts->meta_data_size > 0)
		memcpy(fc_accessor->write_pointer, fc_meta_data,
			   fc_accessor->sts->meta_data_size);
	memcpy(fc_accessor->write_pointer + fc_accessor->sts->meta_data_size, fc_tuple,
		   fc_tuple->t_len);
	fc_accessor->write_pointer += fc_size;
	++fc_accessor->write_chunk->ntuples;
}

static MinimalTuple fc_sts_read_tuple(SharedTuplestoreAccessor *fc_accessor, void *fc_meta_data)
{
	MinimalTuple fc_tuple;
	uint32		fc_size;
	size_t		fc_remaining_size;
	size_t		fc_this_chunk_size;
	char	   *fc_destination;

	/*
	 * 我们将跟踪从这个块读取的字节，以便可以检测到溢出的元组并切换到读取溢出页面。
	 */
	if (fc_accessor->sts->meta_data_size > 0)
	{
		if (BufFileRead(fc_accessor->read_file,
						fc_meta_data,
						fc_accessor->sts->meta_data_size) !=
			fc_accessor->sts->meta_data_size)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from shared tuplestore temporary file"),
					 errdetail_internal("Short read while reading meta-data.")));
		fc_accessor->read_bytes += fc_accessor->sts->meta_data_size;
	}
	if (BufFileRead(fc_accessor->read_file,
					&fc_size,
					sizeof(fc_size)) != sizeof(fc_size))
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read from shared tuplestore temporary file"),
				 errdetail_internal("Short read while reading size.")));
	fc_accessor->read_bytes += sizeof(fc_size);
	if (fc_size > fc_accessor->read_buffer_size)
	{
		size_t		fc_new_read_buffer_size;

		if (fc_accessor->read_buffer != NULL)
			pfree(fc_accessor->read_buffer);
		fc_new_read_buffer_size = Max(fc_size, fc_accessor->read_buffer_size * 2);
		fc_accessor->read_buffer =
			MemoryContextAlloc(fc_accessor->context, fc_new_read_buffer_size);
		fc_accessor->read_buffer_size = fc_new_read_buffer_size;
	}
	fc_remaining_size = fc_size - sizeof(uint32);
	fc_this_chunk_size = Min(fc_remaining_size,
						  BLCKSZ * STS_CHUNK_PAGES - fc_accessor->read_bytes);
	fc_destination = fc_accessor->read_buffer + sizeof(uint32);
	if (BufFileRead(fc_accessor->read_file,
					fc_destination,
					fc_this_chunk_size) != fc_this_chunk_size)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read from shared tuplestore temporary file"),
				 errdetail_internal("Short read while reading tuple.")));
	fc_accessor->read_bytes += fc_this_chunk_size;
	fc_remaining_size -= fc_this_chunk_size;
	fc_destination += fc_this_chunk_size;
	++fc_accessor->read_ntuples;

	/* 检查我们是否需要读取任何溢出块。 */
	while (fc_remaining_size > 0)
	{
		/* 我们现在位于溢出块的起始位置。 */
		SharedTuplestoreChunk fc_chunk_header;

		if (BufFileRead(fc_accessor->read_file, &fc_chunk_header, STS_CHUNK_HEADER_SIZE) !=
			STS_CHUNK_HEADER_SIZE)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from shared tuplestore temporary file"),
					 errdetail_internal("Short read while reading overflow chunk header.")));
		fc_accessor->read_bytes = STS_CHUNK_HEADER_SIZE;
		if (fc_chunk_header.overflow == 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("unexpected chunk in shared tuplestore temporary file"),
					 errdetail_internal("Expected overflow chunk.")));
		fc_accessor->read_next_page += STS_CHUNK_PAGES;
		fc_this_chunk_size = Min(fc_remaining_size,
							  BLCKSZ * STS_CHUNK_PAGES -
							  STS_CHUNK_HEADER_SIZE);
		if (BufFileRead(fc_accessor->read_file,
						fc_destination,
						fc_this_chunk_size) != fc_this_chunk_size)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from shared tuplestore temporary file"),
					 errdetail_internal("Short read while reading tuple.")));
		fc_accessor->read_bytes += fc_this_chunk_size;
		fc_remaining_size -= fc_this_chunk_size;
		fc_destination += fc_this_chunk_size;

		/*
		 * 这些将用于计算在溢出块中溢出的常规元组之后的常规元组。
		 */
		fc_accessor->read_ntuples = 0;
		fc_accessor->read_ntuples_available = fc_chunk_header.ntuples;
	}

	fc_tuple = (MinimalTuple) fc_accessor->read_buffer;
	fc_tuple->t_len = fc_size;

	return fc_tuple;
}

/*
 * 获取当前并行扫描中的下一个元组。
 */
MinimalTuple sts_parallel_scan_next(SharedTuplestoreAccessor *fc_accessor, void *fc_meta_data)
{
	SharedTuplestoreParticipant *fc_p;
	BlockNumber fc_read_page;
	bool		fc_eof;

	for (;;)
	{
		/* 我们可以从当前块中读取更多元组吗？ */
		if (fc_accessor->read_ntuples < fc_accessor->read_ntuples_available)
			return fc_sts_read_tuple(fc_accessor, fc_meta_data);

		/* 找到要读取的新块的位置。 */
		fc_p = &fc_accessor->sts->participants[fc_accessor->read_participant];

		LWLockAcquire(&fc_p->lock, LW_EXCLUSIVE);
		/* 我们可以直接跳过已知的溢出页面。 */
		if (fc_p->read_page < fc_accessor->read_next_page)
			fc_p->read_page = fc_accessor->read_next_page;
		fc_eof = fc_p->read_page >= fc_p->npages;
		if (!fc_eof)
		{
			/* 声明下一个块。 */
			fc_read_page = fc_p->read_page;
			/* 为下一个读取者推进读取头。 */
			fc_p->read_page += STS_CHUNK_PAGES;
			fc_accessor->read_next_page = fc_p->read_page;
		}
		LWLockRelease(&fc_p->lock);

		if (!fc_eof)
		{
			SharedTuplestoreChunk fc_chunk_header;
			size_t		fc_nread;

			/* 确保我们已打开文件。 */
			if (fc_accessor->read_file == NULL)
			{
				char		fc_name[MAXPGPATH];

				fc_sts_filename(fc_name, fc_accessor, fc_accessor->read_participant);
				fc_accessor->read_file =
					BufFileOpenFileSet(&fc_accessor->fileset->fs, fc_name, O_RDONLY,
									   false);
			}

			/* 定位并加载块头。 */
			if (BufFileSeekBlock(fc_accessor->read_file, fc_read_page) != 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not seek to block %u in shared tuplestore temporary file",
								fc_read_page)));
			fc_nread = BufFileRead(fc_accessor->read_file, &fc_chunk_header,
								STS_CHUNK_HEADER_SIZE);
			if (fc_nread != STS_CHUNK_HEADER_SIZE)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not read from shared tuplestore temporary file: read only %zu of %zu bytes",
								fc_nread, STS_CHUNK_HEADER_SIZE)));

			/*
			 * 如果这是一个溢出块，我们会跳过它以及之后的
			 * 所有溢出块。
			 */
			if (fc_chunk_header.overflow > 0)
			{
				fc_accessor->read_next_page = fc_read_page +
					fc_chunk_header.overflow * STS_CHUNK_PAGES;
				continue;
			}

			fc_accessor->read_ntuples = 0;
			fc_accessor->read_ntuples_available = fc_chunk_header.ntuples;
			fc_accessor->read_bytes = STS_CHUNK_HEADER_SIZE;

			/* 再次循环，这样我们就可以从这个块中获取一个元组。 */
		}
		else
		{
			if (fc_accessor->read_file != NULL)
			{
				BufFileClose(fc_accessor->read_file);
				fc_accessor->read_file = NULL;
			}

			/*
			 * 尝试下一个参与者的文件。如果我们已经满圈，
			 * 那我们就完成了。
			 */
			fc_accessor->read_participant = (fc_accessor->read_participant + 1) %
				fc_accessor->sts->nparticipants;
			if (fc_accessor->read_participant == fc_accessor->participant)
				break;
			fc_accessor->read_next_page = 0;

			/* 再次循环，这样我们就可以从这个文件中获取一个块。 */
		}
	}

	return NULL;
}

/*
 * 创建用于参与者将写入的 BufFile 的名称。
 */
static void fc_sts_filename(char *fc_name, SharedTuplestoreAccessor *fc_accessor, int fc_participant)
{
	snprintf(fc_name, MAXPGPATH, "%s.p%d", fc_accessor->sts->name, fc_participant);
}
